package task

import (
	"context"
	"errors"
	"fmt"
	"maps"
	"time"

	"gitee.com/flycash/distributed_task_platform/internal/errs"
	"gitee.com/flycash/distributed_task_platform/internal/event"
	"gitee.com/flycash/distributed_task_platform/internal/service/acquirer"
	"gitee.com/flycash/distributed_task_platform/internal/service/invoker"
	"gitee.com/flycash/distributed_task_platform/pkg/grpc/registry"
	"gitee.com/flycash/distributed_task_platform/pkg/grpc/registry/etcd"
	"gitee.com/flycash/distributed_task_platform/pkg/retry"
	"github.com/ecodeclub/ekit/slice"
	"github.com/gotomicro/ego/core/elog"
	"go.uber.org/multierr"

	"gitee.com/flycash/distributed_task_platform/internal/domain"
	"gitee.com/flycash/distributed_task_platform/internal/repository"
)

// ExecutionService 任务执行服务接口
type ExecutionService interface {
	// Create 创建任务执行实例
	Create(ctx context.Context, execution domain.TaskExecution) (domain.TaskExecution, error)
	// CreateShardingChildren 创建分片子任务执行实例
	CreateShardingChildren(ctx context.Context, parent domain.TaskExecution) ([]domain.TaskExecution, error)
	// FindByID 根据ID获取执行实例
	FindByID(ctx context.Context, id int64) (domain.TaskExecution, error)
	// FindRetryableExecutions 查找所有可以重试的执行记录
	// limit: 查询结果数量限制
	FindRetryableExecutions(ctx context.Context, limit int) ([]domain.TaskExecution, error)
	// FindReschedulableExecutions 查找所有可以重调度的执行记录
	FindReschedulableExecutions(ctx context.Context, limit int) ([]domain.TaskExecution, error)
	// FindShardingParents 查找分片父任务
	FindShardingParents(ctx context.Context, offset, batchSize int) ([]domain.TaskExecution, error)
	// FindShardingChildren 查找分片子任务
	FindShardingChildren(ctx context.Context, parentID int64) ([]domain.TaskExecution, error)
	FindExecutionByTaskIDAndPlanExecID(ctx context.Context, taskID int64, planExecID int64) (domain.TaskExecution, error)
	// FindTimeoutExecutions 查找超时的执行记录
	FindTimeoutExecutions(ctx context.Context, limit int) ([]domain.TaskExecution, error)

	// SetRunningState 设置任务为运行状态并更新进度
	SetRunningState(ctx context.Context, id int64, progress int32, executorNodeID string) error
	// UpdateRunningProgress 更新任务执行进度（仅在RUNNING状态下有效）
	UpdateRunningProgress(ctx context.Context, id int64, progress int32) error
	// UpdateRetryResult 更新重试结果
	UpdateRetryResult(ctx context.Context, id, retryCount, nextRetryTime int64, status domain.TaskExecutionStatus, progress int32, endTime int64, scheduleParams map[string]string, executorNodeID string) error
	// UpdateScheduleResult 更新调度结果
	UpdateScheduleResult(ctx context.Context, id int64, status domain.TaskExecutionStatus, progress int32, endTime int64, scheduleParams map[string]string, executorNodeID string) error

	// HandleReports 处理执行节点上报的执行状态
	HandleReports(ctx context.Context, reports []*domain.Report) error
	// UpdateState 更新执行节点上报的执行状态
	UpdateState(ctx context.Context, state domain.ExecutionState) error
}

type executionService struct {
	nodeID       string
	repo         repository.TaskExecutionRepository
	taskSvc      Service
	taskAcquirer acquirer.TaskAcquirer  // 任务抢占器
	producer     event.CompleteProducer // 任务完成事件生产者
	registry     registry.Registry
	invoker      invoker.Invoker
	logger       *elog.Component
}

// NewExecutionService 创建任务执行服务实例
func NewExecutionService(
	nodeID string,
	repo repository.TaskExecutionRepository,
	taskSvc Service,
	taskAcquirer acquirer.TaskAcquirer,
	producer event.CompleteProducer,
	registry *etcd.Registry,
	invoker invoker.Invoker,
) ExecutionService {
	return &executionService{
		nodeID:       nodeID,
		repo:         repo,
		taskSvc:      taskSvc,
		taskAcquirer: taskAcquirer,
		producer:     producer,
		registry:     registry,
		invoker:      invoker,
		logger:       elog.DefaultLogger.With(elog.FieldComponentName("service.execution")),
	}
}

func (s *executionService) Create(ctx context.Context, execution domain.TaskExecution) (domain.TaskExecution, error) {
	return s.repo.Create(ctx, execution)
}

func (s *executionService) CreateShardingChildren(ctx context.Context, parent domain.TaskExecution) ([]domain.TaskExecution, error) {
	if parent.Task.ID == 0 {
		return nil, errors.New("Task.ID不能为空")
	}
	shardingRule := parent.Task.ShardingRule
	if shardingRule == nil {
		return nil, errs.ErrTaskShardingRuleNotFound
	}

	var executorNodeIDs []string

	if shardingRule.Type.IsWeightedDynamicRange() {
		// 发起Prepare调用
		params, err := s.invoker.Prepare(ctx, parent)
		if err != nil {
			return nil, fmt.Errorf("发送GRPC请求失败: %w", err)
		}
		// 将获取到的返回参数合并到分片规则中
		maps.Copy(shardingRule.Params, params)

		// 调用Registry组件获取，执行节点信息列表（信息内含有权重）
		serviceInstances, err := s.registry.ListServices(ctx, parent.Task.GrpcConfig.ServiceName)
		if err != nil {
			return nil, fmt.Errorf("获取执行节点注册信息失败: %w", err)
		}
		if len(serviceInstances) == 0 {
			return nil, fmt.Errorf("未找到任何可用的执行节点")
		}
		// 将执行节点的信息注入分片规则中
		shardingRule.ExecutorNodeInstances = serviceInstances
		// 按照顺序获取执行节点ID
		executorNodeIDs = slice.Map(serviceInstances, func(_ int, src registry.ServiceInstance) string {
			return src.ID
		})
	}

	// 计算分片任务需要的分片调度参数
	scheduleParams := shardingRule.ToScheduleParams()

	s.logger.Info("构建分片规则调度参数成功",
		elog.Any("executorNodeIDs", executorNodeIDs),
		elog.Any("scheduleParams", scheduleParams))

	// 创建父任务执行记录
	parent.Status = domain.TaskExecutionStatusRunning
	created, err := s.repo.CreateShardingParent(ctx, parent)
	if err != nil {
		return nil, err
	}
	// 这里有一个要求： executorNodeIDs[i] 与 scheduleParams[i] 是对应关系
	return s.repo.CreateShardingChildren(ctx, created, executorNodeIDs, scheduleParams)
}

func (s *executionService) FindByID(ctx context.Context, id int64) (domain.TaskExecution, error) {
	return s.repo.GetByID(ctx, id)
}

func (s *executionService) FindRetryableExecutions(ctx context.Context, limit int) ([]domain.TaskExecution, error) {
	return s.repo.FindRetryableExecutions(ctx, limit)
}

func (s *executionService) FindReschedulableExecutions(ctx context.Context, limit int) ([]domain.TaskExecution, error) {
	return s.repo.FindReschedulableExecutions(ctx, limit)
}

func (s *executionService) FindShardingParents(ctx context.Context, offset, batchSize int) ([]domain.TaskExecution, error) {
	return s.repo.FindShardingParents(ctx, offset, batchSize)
}

func (s *executionService) FindShardingChildren(ctx context.Context, parentID int64) ([]domain.TaskExecution, error) {
	return s.repo.FindShardingChildren(ctx, parentID)
}

func (s *executionService) FindExecutionByTaskIDAndPlanExecID(ctx context.Context, taskID, planExecID int64) (domain.TaskExecution, error) {
	return s.repo.FindExecutionByTaskIDAndPlanExecID(ctx, taskID, planExecID)
}

func (s *executionService) FindTimeoutExecutions(ctx context.Context, limit int) ([]domain.TaskExecution, error) {
	return s.repo.FindTimeoutExecutions(ctx, limit)
}

func (s *executionService) SetRunningState(ctx context.Context, id int64, progress int32, executorNodeID string) error {
	return s.repo.SetRunningState(ctx, id, progress, executorNodeID)
}

func (s *executionService) UpdateRunningProgress(ctx context.Context, id int64, progress int32) error {
	return s.repo.UpdateRunningProgress(ctx, id, progress)
}

func (s *executionService) UpdateRetryResult(ctx context.Context, id, retryCount, nextRetryTime int64, status domain.TaskExecutionStatus, progress int32, endTime int64, scheduleParams map[string]string, executorNodeID string) error {
	return s.repo.UpdateRetryResult(ctx, id, retryCount, nextRetryTime, status, progress, endTime, scheduleParams, executorNodeID)
}

func (s *executionService) UpdateScheduleResult(ctx context.Context, id int64, status domain.TaskExecutionStatus, progress int32, endTime int64, scheduleParams map[string]string, executorNodeID string) error {
	return s.repo.UpdateScheduleResult(ctx, id, status, progress, endTime, scheduleParams, executorNodeID)
}

func (s *executionService) HandleReports(ctx context.Context, reports []*domain.Report) error {
	if len(reports) == 0 {
		return nil
	}
	s.logger.Debug("开始处理执行状态上报", elog.Int("count", len(reports)))

	var err error
	processedCount := 0
	skippedCount := 0

	for i := range reports {
		err1 := s.UpdateState(ctx, reports[i].ExecutionState)
		if err1 != nil {
			skippedCount++
			s.logger.Error("处理执行节点上报的结果失败",
				elog.Any("result", reports[i].ExecutionState),
				elog.FieldErr(err1))
			// 包装错误，添加上报场景的特定信息
			err = multierr.Append(err,
				fmt.Errorf("处理执行节点上报的结果失败: taskID=%d, executionID=%d: %w",
					reports[i].ExecutionState.TaskID, reports[i].ExecutionState.ID, err1))
			continue
		}
		processedCount++
	}

	// 记录处理统计信息
	s.logger.Info("执行状态上报处理完成",
		elog.Int("total", len(reports)),
		elog.Int("processed", processedCount),
		elog.Int("skipped", skippedCount))
	return err
}

func (s *executionService) UpdateState(ctx context.Context, state domain.ExecutionState) error {
	execution, err := s.FindByID(ctx, state.ID)
	if err != nil {
		return errs.ErrExecutionNotFound
	}

	// 已处于终止状态的的执行记录不允许再进行状态迁移
	if execution.Status.IsTerminalStatus() {
		s.logger.Error("错乱的状态迁移",
			elog.Int64("taskID", execution.Task.ID),
			elog.String("taskName", execution.Task.Name),
			elog.String("currentStatus", execution.Status.String()),
			elog.String("targetStatus", state.Status.String()))
		return errs.ErrInvalidTaskExecutionStatus
	}

	switch {
	case state.Status.IsRunning():
		if execution.Status.IsRunning() {
			// 仅更新进度
			return s.updateRunningProgress(ctx, state)
		}
		// 设置为RUNNING状态的同时设置开始时间
		return s.setRunningState(ctx, state)
	case state.Status.IsFailedRetryable():
		err = s.updateRetryState(ctx, execution, state)
		if err != nil {
			s.logger.Error("更新任务执行记录的重试结果失败",
				elog.Int64("taskID", state.TaskID),
				elog.String("taskName", state.TaskName),
				elog.Any("state", state),
				elog.FieldErr(err))

			// 达到最大重试次数
			if errors.Is(err, errs.ErrExecutionMaxRetriesExceeded) {
				// 发送完成事件
				s.sendCompletedEvent(ctx, state, execution)
			}
			return err
		}
		return nil
	case state.Status.IsFailedRescheduled():
		if state.RequestReschedule {
			// 更新调度信息
			execution.MergeTaskScheduleParams(state.RescheduleParams)
		}
		err = s.updateState(ctx, execution, state)
		if err != nil {
			return fmt.Errorf("更新任务执行记录的重调度结果失败：%w", err)
		}
		return nil
	case state.Status.IsTerminalStatus():
		var err1 error
		if err2 := s.updateState(ctx, execution, state); err2 != nil {
			err1 = multierr.Append(err1, fmt.Errorf("更新任务执行记录的调度结果失败：%w", err2))
		}
		// 不管本次调度是否成功，都要更新task的下一次执行时间，分片任务除外，分片任务会在补偿任务重进行
		isShardedTask := execution.ShardingParentID != nil && *execution.ShardingParentID > 0
		if !isShardedTask {
			// 释放任务
			s.releaseTask(ctx, execution.Task)
			// 更新任务执行
			if _, err3 := s.taskSvc.UpdateNextTime(ctx, execution.Task.ID); err3 != nil {
				err1 = multierr.Append(err1, fmt.Errorf("更新任务下次更新时间失败：%w", err3))
			}
		}
		// 发送完成事件
		s.sendCompletedEvent(ctx, state, execution)
		return err1
	default:
		s.logger.Error("非法上报状态",
			elog.Int64("taskID", execution.Task.ID),
			elog.String("taskName", execution.Task.Name),
			elog.String("currentStatus", execution.Status.String()),
			elog.String("targetStatus", state.Status.String()))
		return errs.ErrInvalidTaskExecutionStatus
	}
}

func (s *executionService) updateRunningProgress(ctx context.Context, state domain.ExecutionState) error {
	err := s.UpdateRunningProgress(ctx, state.ID, state.RunningProgress)
	if err != nil {
		s.logger.Error("更新运行进度失败",
			elog.Int64("taskID", state.TaskID),
			elog.String("taskName", state.TaskName),
			elog.Any("state", state),
			elog.FieldErr(err))
		return err
	}
	return nil
}

func (s *executionService) setRunningState(ctx context.Context, state domain.ExecutionState) error {
	err := s.SetRunningState(ctx, state.ID, state.RunningProgress, state.ExecutorNodeID)
	if err != nil {
		s.logger.Error("更新为运行状态失败",
			elog.Int64("taskID", state.TaskID),
			elog.String("taskName", state.TaskName),
			elog.Any("state", state),
			elog.FieldErr(err))
		return err
	}
	return nil
}

func (s *executionService) updateRetryState(ctx context.Context, execution domain.TaskExecution, state domain.ExecutionState) error {
	// 计算出下次重试时间
	retryStrategy, _ := retry.NewRetry(execution.Task.RetryConfig.ToRetryComponentConfig())
	duration, shouldRetry := retryStrategy.NextWithRetries(int32(execution.RetryCount + 1))
	if shouldRetry {
		// 当前不是最后一次重试，计算下次重试时间
		execution.NextRetryTime = time.Now().Add(duration).UnixMilli()
		// 增加重试计数
		execution.RetryCount++
	} else if !state.Status.IsTerminalStatus() {
		// 当前是最后一次重试，只要不是终止状态一律设置为失败
		state.Status = domain.TaskExecutionStatusFailed
	}
	// 不管是否达到最大重试次数，都要更新状态（主要是重试次数），这样下次重试补偿任务会因其超过最大重试次数而不再重试
	err := s.UpdateRetryResult(ctx,
		state.ID,
		execution.RetryCount,
		execution.NextRetryTime,
		state.Status,
		state.RunningProgress,
		time.Now().UnixMilli(),
		execution.Task.ScheduleParams,
		state.ExecutorNodeID)
	if err != nil {
		s.logger.Error("更新执行计划重试结果失败",
			elog.Int64("taskID", execution.Task.ID),
			elog.String("taskName", execution.Task.Name),
			elog.Any("result", state),
			elog.FieldErr(err))
		if !shouldRetry {
			return fmt.Errorf("%w: %w", errs.ErrExecutionMaxRetriesExceeded, err)
		}
		return err
	}
	if !shouldRetry {
		return errs.ErrExecutionMaxRetriesExceeded
	}
	s.logger.Info("更新重试状态成功",
		elog.Int64("taskID", execution.Task.ID),
		elog.String("taskName", execution.Task.Name),
		elog.Any("state", state))
	return nil
}

func (s *executionService) updateState(ctx context.Context, execution domain.TaskExecution, state domain.ExecutionState) error {
	err := s.UpdateScheduleResult(ctx,
		state.ID,
		state.Status,
		state.RunningProgress,
		time.Now().UnixMilli(),
		execution.Task.ScheduleParams,
		state.ExecutorNodeID)
	if err != nil {
		s.logger.Error("更新调度结果失败",
			elog.Int64("taskID", execution.Task.ID),
			elog.String("taskName", execution.Task.Name),
			elog.Any("state", state),
			elog.FieldErr(err))
		return err
	}
	s.logger.Info("更新调度状态成功",
		elog.Int64("taskID", execution.Task.ID),
		elog.String("taskName", execution.Task.Name),
		elog.Any("state", state))
	return nil
}

func (s *executionService) releaseTask(ctx context.Context, task domain.Task) {
	if err := s.taskAcquirer.Release(ctx, task.ID, s.nodeID); err != nil {
		s.logger.Error("释放任务失败",
			elog.Int64("taskID", task.ID),
			elog.String("taskName", task.Name),
			elog.FieldErr(err))
	}
}

func (s *executionService) sendCompletedEvent(ctx context.Context, state domain.ExecutionState, execution domain.TaskExecution) {
	if !state.Status.IsTerminalStatus() {
		// 非终止状态不用做处理
		return
	}
	err := s.producer.Produce(ctx, event.Event{
		PlanID: execution.Task.PlanID,
		TaskID: execution.Task.ID,
		Name:   execution.Task.Name,
		Type:   domain.NormalTaskType,
	})
	if err != nil {
		s.logger.Error("发送完成事件失败", elog.Int64("taskID", execution.Task.ID), elog.FieldErr(err))
	}
}
